//package com.hexiang.consumer;
//
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.apache.rocketmq.remoting.common.RemotingHelper;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.PostConstruct;
//import javax.annotation.PreDestroy;
//
//@Slf4j
//@Component
//public class MockReConsumeByExceptionConsumer {
//
//    private static final Integer MAX_RETRY_TIMES = 2;
//
//    private String consumerGroup = "mock-reconsume-by-exception-consumer-group";
//    private String namesrvAddr = "127.0.0.1:9876";
//    private String topicName = "test_topic";
//
//    private DefaultMQPushConsumer consumer;
//
//    @PostConstruct
//    public void init() throws MQClientException {
//        consumer = consumerInit();
//        consumer.start();
//    }
//
//    @PreDestroy
//    public void shutdown() {
//        consumer.shutdown();
//    }
//
//    private DefaultMQPushConsumer consumerInit() throws MQClientException {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
//        consumer.setNamesrvAddr(namesrvAddr);
//        consumer.setMessageModel(MessageModel.CLUSTERING);
//        // 消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//        // 订阅主题和 标签（ * 代表所有标签)下信息
//        consumer.subscribe(topicName, "*");
//        // 注册消费的监听 并在此监听中消费信息, 并返回消费的状态信息
//        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
//            // msgs 中只收集同一个topic, 同一个 tag, 并且 key 相同的message
//            // 会把不同的消息分别放置到不同的队列中
//            for (MessageExt messageExt : msgs) {
//                try {
//                    //逐条消费
//                    String messageBody = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
//                    log.info("Message Consumer: Handle New Message: messageId: " + messageExt.getMsgId() + ",topic: " +
//                            messageExt.getTopic() + ",tags: " + messageExt.getTags() + ",messageBody: " + messageBody);
//
//                    //模拟业务异常
//                    int i = 1 / 0;
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                } catch (Exception e) {
//                    log.error("Consume Message Error!!", e);
//                    //抛出异常时，返回ConsumeConcurrentlyStatus.RECONSUME_LATER，尝试重试。当重试指定次数后返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS
//                    int reconsumeTimes = messageExt.getReconsumeTimes();
//                    log.error("Now Retry Times: " + reconsumeTimes);
//                    if (reconsumeTimes >= MAX_RETRY_TIMES) {
//                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                    }
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                }
//            }
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        });
//        return consumer;
//    }
//}
